package com.fs.config;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;

import java.util.HashMap;
import java.util.Map;

/**
 * @author huyunqiang
 * @Date 2019/11/23 19:05
 */
@Configuration
@EnableKafka
public class KafkaProducerConfig {
    @Autowired
    PropsConfig propsConfig;

    /**
     * 使用@Transactional注解方式使用kafka 事务
     * kafka配置事务管理 加Transactional.id的前缀。
     * @return
     */
//    @Bean
//    public ProducerFactory<String, String> producerFactory() {
//        DefaultKafkaProducerFactory factory = new DefaultKafkaProducerFactory<>(procuderConfigs());
//        factory.transactionCapable();
//        factory.setTransactionIdPrefix("tran-");
//        return factory;
//    }
//    @Bean
//    public KafkaTransactionManager transactionManager(ProducerFactory producerFactory) {
//        KafkaTransactionManager manager = new KafkaTransactionManager(producerFactory);
//        return manager;
//    }
//
//    @Bean
//    public KafkaTemplate<String, String> kafkaTemplate() {
//        return new KafkaTemplate<String, String>(producerFactory());
//    }


    @Bean
    public Map<String, Object> procuderConfigs() {
        Map<String, Object> propsMap = new HashMap<>();
        propsMap.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, propsConfig.getBroker());
        propsMap.put(ProducerConfig.BATCH_SIZE_CONFIG, "16384");
        propsMap.put(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432");
        propsMap.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, propsConfig.getKey());
        propsMap.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, propsConfig.getValue());
        propsMap.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, propsConfig.getValue());
        //自定义分区器
//        propsMap.put("partitioner.class", "com.fs.config.DefinePartition");
//      自定义生产这拦截器
        propsMap.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, "com.fs.config.ProducerinterceptorPrefix");
        return propsMap;
    }

}
